from kafka import KafkaProducer
from kafka import KafkaConsumer
import json
import pymysql
import yaml


file_path = "../../etc/config.yaml"
kafka_server = ''
kafka_topic = ''

mysql_host = ''
mysql_port = ''
mysql_user = ''
mysql_password = ''
mysql_database = ''
with open(file_path, 'r', encoding='utf-8') as file:
    config = yaml.safe_load(file)
    kafka_config = config.get('kafka', {})
    kafka_server = kafka_config.get('server', '')
    kafka_topic = kafka_config.get('topic', '')

    mysql_config = config.get('mysql', {})
    mysql_host = mysql_config.get('host', '')
    mysql_port = mysql_config.get('port', '')
    mysql_user = mysql_config.get('user', '')
    mysql_password = mysql_config.get('password', '')
    mysql_database = mysql_config.get('database', '')


"""
create table user_follow(
    id int primary key auto_increment,
    follower_id int not null,
    followed_id int not null,
    created_at timestamp default current_timestamp
)
"""

# 初始化 MySQL 连接
mysql_connection = pymysql.connect(
    host=mysql_host,
    port=mysql_port,
    user=mysql_user,
    password=mysql_password,
    database=mysql_database,
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor,
)

follow_table = 'user_follow'

# Kafka 消费者函数，消费消息并写入 MySQL
def consume_and_write_to_mysql():
    consumer = KafkaConsumer(
        kafka_topic,
        bootstrap_servers=kafka_server,
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        group_id='user_follow_consumer_group',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )

    while True:
        for message in consumer:
            data = message.value
            action = data.get("action")
            follower_id = data.get("follower_id")
            followed_id = data.get("followed_id")
            sql = ""
            if action == "user_follow":
                sql = "INSERT INTO {} (follower_id, followed_id) VALUES (%s, %s)".format(follow_table)
            elif action == "user_unfollow":
                sql = "DELETE FROM {} WHERE follower_id = %s AND followed_id = %s".format(follow_table)

            if sql:
                try:
                    with mysql_connection.cursor() as cursor:
                        cursor.execute(sql, (follower_id, followed_id))
                    mysql_connection.commit()
                    print("mysql execute success")
                except Exception as e:
                    print(f"Error executing MySQL query: {e}")


if __name__ == "__main__":
    consume_and_write_to_mysql()